package cn.edu.laud.sany.datatest;

import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.SQLTransientConnectionException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.cassandra.cql.jdbc.CassandraResultSet;
import org.apache.cassandra.cql.jdbc.JdbcUTF8;
import org.apache.cassandra.thrift.TimedOutException;

import cn.edu.laud.sany.utils.MyLog;
import cn.edu.laud.sany.utils.MyResultLog;

/**
 * test runner for import/read using jdbc
 * 
 * @author zhongyu
 * 
 */
public class LaUDTest_JDBC extends Thread
{
	public static int JDBC_BATCH_SIZE = 10000;
	protected static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
	public static ConcurrentLinkedQueue<String> sqlQueue;
	
	protected int type;
	protected MyLog logger;
	protected MyResultLog resultLogger;
	protected Statement laudStatement;
	private Connection laudConnection;
	List<String> sqls = new ArrayList<String>();
	protected int no;
	protected String jdbcString;
	protected String keyspace;
	protected long now = 0;
	protected boolean isImportSucceed = true;
	public static Long testNumber = 1L;
	
	public LaUDTest_JDBC(int no, String jdbcString, String keyspace, String logFileName,
			int testType)
	{
		this.type = testType;
		this.no = no;
		this.jdbcString = jdbcString;
		this.keyspace = keyspace;
		logger = MyLog.getMyLog(logFileName, true);
		logger.setThreadName("Thread " + no);
	}
	
	protected void closeLaUD()
	{
		if (laudStatement != null)
		{
			try
			{
				laudStatement.close();
			}
			catch (SQLException e)
			{
				e.printStackTrace();
				logger.error(e.getMessage());
			}
		}
		if (laudConnection != null)
		{
			try
			{
				laudConnection.close();
			}
			catch (SQLException e)
			{
				e.printStackTrace();
				logger.error(e.getMessage());
			}
		}
	}
	
	public boolean beginConnection()
	{
		try
		{
			try
			{
				logger.info("try get jdbc connection...");
				laudConnection = DriverManager.getConnection(jdbcString, "", "");
				laudStatement = laudConnection.createStatement();
				logger.info("get jdbc connection successful.");
			}
			catch (SQLException e1)
			{
				logger.error("get jdbc connection error:" + e1.getMessage() + ", jdbc url:"
						+ jdbcString);
				return false;
			}
			laudStatement.execute("use " + keyspace);
		}
		catch (SQLException e1)
		{
			logger.error(e1.getMessage());
			return false;
		}
		return true;
	}
	
	public void closeDB()
	{
		try
		{
			closeLaUD();
			logger.info("laud connect closed successfully.");
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	
	@Override
	public void run()
	{
		if (type == 0)
		{
			run_Test_Import();
		}
		else
		{
			run_Test_Read(type);
		}
	}
	
	public void begin_Test_Import()
	{
		try
		{
			while (true)
			{
				if (beginConnection() == false)
				{
					logger.info("begin import failed, thread will sleep 5 sec to restart...");
					try
					{
						Thread.sleep(5000);
					}
					catch (InterruptedException e)
					{
						e.printStackTrace();
					}
				}
				else
				{
					break;
				}
			}
			
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	
	public void run_Test_Import()
	{
		try
		{
			// logger.info(String.format("import start...", no));
			// import
			importData();
			// logger.info(String.format("import finished!"));
			
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	
	private void importData()
	{
		while (true)
		{
			if (sqlQueue.size() == 0)
			{
				return;
			}
			String sqlString = null;
			synchronized (sqlQueue)
			{
				if (sqlQueue.size() > 0)
				{
					sqlString = sqlQueue.poll();
				}
			}
			if (sqlString == null)
			{
				continue;
			}
			executeSQL(sqlString);
		}
	}
	
	protected boolean executeSQL(String sqls)
	{
		int timeouttime = 5;
		while (timeouttime > 0)
		{
			try
			{
				laudStatement.execute(sqls);
				return true;
			}
			catch (SQLException e)
			{
				logger.equals(e.getMessage());
				if (timeouttime > 0 && (e.getCause() instanceof TimedOutException))
				{
					logger.warn("time out exception, will try again after 5 secondes,[time "
							+ timeouttime + "]");
					timeouttime--;
					try
					{
						Thread.sleep(5000);
					}
					catch (InterruptedException e1)
					{
						logger.equals(e.getMessage());
					}
				}
				else if (timeouttime > 0
						&& (e.getCause() instanceof SQLTransientConnectionException))
				{
					logger.warn(e.getMessage() + ", will try again after 10 secondes,[time "
							+ timeouttime + "]");
					timeouttime--;
					try
					{
						Thread.sleep(10000);
					}
					catch (InterruptedException e1)
					{
						logger.equals(e.getMessage());
					}
				}
				else
				{
					logger.error(e.getMessage());
					return false;
				}
			}
		}
		return false;
	}
	
	public void run_Test_Read(int type)
	{
		try
		{
			while (true)
			{
				if (beginConnection() == false)
				{
					logger.info("get connection failed, thread will sleep 5 sec to restart...");
					try
					{
						Thread.sleep(5000);
					}
					catch (InterruptedException e)
					{
						e.printStackTrace();
					}
				}
				else
				{
					break;
				}
			}
			// test
			if (type == 1)
			{
				testRead();
			}
			else if (type == 2)
			{
				testReadHot_1();
			}
			else if (type == 3)
			{
				testReadHot_2();
			}
			else if (type == 4)
			{
				testRemind2();
			}
			
			logger.info(String.format("test finished!"));
			
		}
		catch (Exception e)
		{
			logger.error(e.getMessage());
			e.printStackTrace();
		}
		finally
		{
			closeDB();
		}
	}
	
	private void testRead()
	{
		resultLogger = MyResultLog.getMyLog(
				LaUDReadTest.getLogFilePath(LaUDReadTest.resultLogFileName), false);
		Random ran = new Random();
		int cfsCount = LaUDReadTest.cFS.length;
		int equipCount = LaUDReadTest.equips.length;
		int userGiveTimeCount = LaUDReadTest.startQueryTimes.length;
		for (int i = 0; i < LaUDReadTest.testRepeatNumber; i++)
		{
			int paraId = LaUDReadTest.cFS[ran.nextInt(cfsCount)];
			String equipId = LaUDReadTest.equips[ran.nextInt(equipCount)];
			long startTime, endTime;
			Date startdate, enddate;
			int queryDays = 0;
			if (i < userGiveTimeCount)
			{
				startdate = LaUDReadTest.startQueryTimes[i];
				enddate = LaUDReadTest.endQueryTimes[i];
				startTime = startdate.getTime();
				endTime = enddate.getTime();
				queryDays = calInterval(startdate, enddate);
			}
			else
			{
				int allDaysInterval = calInterval(LaUDReadTest.startGetQueryDate,
						LaUDReadTest.endGetQueryDate);
				Calendar calendar = Calendar.getInstance();
				calendar.setTime(LaUDReadTest.startGetQueryDate);
				calendar.add(Calendar.DAY_OF_YEAR, ran.nextInt(allDaysInterval / 2));
				startdate = calendar.getTime();
				startTime = startdate.getTime();
				queryDays = ran.nextInt(allDaysInterval / 3);
				calendar.add(Calendar.DAY_OF_YEAR, queryDays);
				enddate = calendar.getTime();
				endTime = enddate.getTime();
			}
			long startColumn, endColumn;
			if (LaUDReadTest.columnReversed)
			{
				startColumn = endTime;
				endColumn = startTime;
			}
			else
			{
				startColumn = startTime;
				endColumn = endTime;
			}
			String selectTimeString = String.valueOf(startColumn) + ".."
					+ String.valueOf(endColumn);
			String cql = "";
			if (LaUDReadTest.schema == -1)
			{
				cql = String.format(LaUDReadTest.cqlSelectFormatString2, selectTimeString, equipId,
						paraId);
			}
			else
			{
				cql = String.format(LaUDReadTest.cqlSelectFormatString, selectTimeString, paraId,
						equipId);
			}
			long startTestTime = System.nanoTime();
			int result = executeSelectStatment(cql, true);
			long endTestTime = System.nanoTime();
			String logString = buildResultLog(startTestTime, paraId, equipId, startdate, enddate,
					queryDays, result, endTestTime);
			resultLogger.log(logString);
		}
	}
	
	private void testReadHot_1()
	{
		resultLogger = MyResultLog.getMyLog(
				LaUDReadTest.getLogFilePath(LaUDReadTest.resultLogFileName), false);
		Random ran = new Random();
		int cfsCount = LaUDReadTest.cFS.length;
		int equipCount = LaUDReadTest.equips.length;
		int queryWhereCount = LaUDReadTest.querySubCount;
		if (queryWhereCount > equipCount)
		{
			queryWhereCount = equipCount;
		}
		for (int i = 0; i < LaUDReadTest.testRepeatNumber; i++)
		{
			int paraId = LaUDReadTest.cFS[ran.nextInt(cfsCount)];
			List<String> equipIds = new ArrayList<String>();
			while (equipIds.size() < queryWhereCount)
			{
				String equipId = LaUDReadTest.equips[ran.nextInt(equipCount)];
				if (equipIds.contains(equipId) == false)
				{
					equipIds.add(equipId);
				}
			}
			StringBuilder sb = new StringBuilder();
			sb.append("equip in ('");
			StringBuilder sb2 = new StringBuilder();
			for (int j = 0; j < equipIds.size() - 1; j++)
			{
				sb.append(equipIds.get(j));
				sb.append("', '");
				sb2.append(equipIds.get(j));
				sb2.append(";");
			}
			sb.append(equipIds.get(equipIds.size() - 1));
			sb.append("')");
			sb2.append(equipIds.get(equipIds.size() - 1));
			String cql = String.format(LaUDReadTest.cqlSelectWhereFormatString, "First 1 *",
					paraId, sb.toString());
			long startTestTime = System.nanoTime();
			int result = executeSelectStatment(cql, false);
			long endTestTime = System.nanoTime();
			String logString = buildResultLog2(startTestTime, 1, String.valueOf(paraId),
					sb2.toString(), result, endTestTime);
			resultLogger.log(logString);
		}
	}
	
	private void testReadHot_2()
	{
		resultLogger = MyResultLog.getMyLog(
				LaUDReadTest.getLogFilePath(LaUDReadTest.resultLogFileName), false);
		Random ran = new Random();
		int cfsCount = LaUDReadTest.cFS.length;
		int equipCount = LaUDReadTest.equips.length;
		int queryCQLCount = LaUDReadTest.querySubCount;
		if (queryCQLCount > cfsCount)
		{
			queryCQLCount = cfsCount;
		}
		for (int i = 0; i < LaUDReadTest.testRepeatNumber; i++)
		{
			String equipId = LaUDReadTest.equips[ran.nextInt(equipCount)];
			List<Integer> paraIds = new ArrayList<Integer>();
			while (paraIds.size() < queryCQLCount)
			{
				int paraId = LaUDReadTest.cFS[ran.nextInt(cfsCount)];
				if (paraIds.contains(paraId) == false)
				{
					paraIds.add(paraId);
				}
			}
			for (int j = 0; j < paraIds.size(); j++)
			{
				int paraId = paraIds.get(j);
				String cql = String.format(LaUDReadTest.cqlSelectFormatString, "First 1 *", paraId,
						equipId);
				long startTestTime = System.nanoTime();
				int result = executeSelectStatment(cql, false);
				long endTestTime = System.nanoTime();
				String logString = buildResultLog2(startTestTime, 1, equipId,
						String.valueOf(paraId), result, endTestTime);
				resultLogger.log(logString);
			}
		}
	}
	
	private static int[] remindCFs = { 209, 280, 252, 211, 183, 287 };
	
	private void testRemind()
	{
		resultLogger = MyResultLog.getMyLog(
				LaUDReadTest.getLogFilePath(LaUDReadTest.resultLogFileName), false);
		String[] equipIds = LaUDReadTest.equips;
		int equipCount = LaUDReadTest.equips.length;
		int queryWhereCount = LaUDReadTest.querySubCount;
		if (queryWhereCount > equipCount)
		{
			queryWhereCount = equipCount;
		}
		for (int i = 0; i < LaUDReadTest.testRepeatNumber; i++)
		{
			for (int k = 0; k < remindCFs.length; k++)
			{
				int paraId = remindCFs[k];
				StringBuilder sb = new StringBuilder();
				sb.append("key in ('");
				for (int j = 0; j < equipCount - 1; j++)
				{
					sb.append(equipIds[j]);
					sb.append(paraId);
					sb.append("', '");
				}
				sb.append(equipIds[equipCount - 1]);
				sb.append(paraId);
				sb.append("') and column > 0 ");
				String cql = String.format(LaUDReadTest.cqlSelectFormatString3, sb.toString());
				long startTestTime = System.nanoTime();
				int result = executeSelectStatment(cql, false);
				long endTestTime = System.nanoTime();
				String logString = buildResultLog4(startTestTime, String.valueOf(paraId), result,
						endTestTime);
				resultLogger.log(logString);
				
			}
		}
	}
	
	private void testRemind2()
	{
		resultLogger = MyResultLog.getMyLog(
				LaUDReadTest.getLogFilePath(LaUDReadTest.resultLogFileName), false);
		String[] equipIds = LaUDReadTest.equips;
		int equipCount = LaUDReadTest.equips.length;
		int queryWhereCount = LaUDReadTest.querySubCount;
		if (queryWhereCount > equipCount)
		{
			queryWhereCount = equipCount;
		}
		for (int i = 0; i < LaUDReadTest.testRepeatNumber; i++)
		{
			for (int k = 0; k < remindCFs.length; k++)
			{
				long startTestTime = System.nanoTime();
				int paraId = remindCFs[k];
				int result = 0;
				for (int j = 0; j < equipCount - 1; j++)
				{
					StringBuilder sb = new StringBuilder();
					sb.append("key = '");
					sb.append(equipIds[j]);
					sb.append(paraId);
					sb.append("' and value > 0 ");
					String cql = String.format(LaUDReadTest.cqlSelectFormatString3, sb.toString());
					result += executeSelectStatment2(cql);
				}				
				long endTestTime = System.nanoTime();
				String logString = buildResultLog4(startTestTime, String.valueOf(paraId), result,
						endTestTime);
				resultLogger.log(logString);
				
			}
		}
	}
	
	private String buildResultLog(long startTestTime, int paraId, String equipId,
			Date startQueryDate, Date endQueryDate, int queryDays, int result, long endTestTime)
	{
		StringBuilder sb = new StringBuilder();
		synchronized (testNumber)
		{
			sb.append(String.valueOf(testNumber));
			if (testNumber % 1000 == 0)
			{
				logger.info("total query test: " + testNumber);
			}
			testNumber++;
		}
		sb.append(",");
		sb.append(String.valueOf(no));
		sb.append(",");
		double timeused = (double) (endTestTime - startTestTime) / 1000000.0;
		sb.append(String.valueOf(timeused));
		sb.append(",");
		sb.append(paraId);
		sb.append(",");
		sb.append(equipId);
		sb.append(",");
		sb.append(queryDays);
		sb.append(",");
		sb.append(sdf.format(startQueryDate));
		sb.append(",");
		sb.append(sdf.format(endQueryDate));
		sb.append(",");
		sb.append((result != -1));
		sb.append(",");
		sb.append(result);
		String logString = sb.toString();
		return logString;
	}
	
	private String buildResultLog2(long startTestTime, int queryType, String equipIdOrParaId,
			String equipIdOrParaIdList, int result, long endTestTime)
	{
		StringBuilder sb = new StringBuilder();
		synchronized (testNumber)
		{
			sb.append(String.valueOf(testNumber));
			if (testNumber % 1000 == 0)
			{
				logger.info("total query test: " + testNumber);
			}
			testNumber++;
		}
		sb.append(",");
		sb.append(String.valueOf(no));
		sb.append(",");
		double timeused = (double) (endTestTime - startTestTime) / 1000000.0;
		sb.append(String.valueOf(timeused));
		sb.append(",");
		sb.append(queryType);
		sb.append(",");
		sb.append(equipIdOrParaId);
		sb.append(",");
		sb.append(equipIdOrParaIdList);
		sb.append(",");
		sb.append((result != -1));
		sb.append(",");
		sb.append(result);
		String logString = sb.toString();
		return logString;
	}
	
	private String buildResultLog4(long startTestTime, String ParaId, int result, long endTestTime)
	{
		StringBuilder sb = new StringBuilder();
		synchronized (testNumber)
		{
			sb.append(String.valueOf(testNumber));
			logger.info("total query test: " + testNumber);
			testNumber++;
		}
		sb.append(",");
		sb.append(String.valueOf(no));
		sb.append(",");
		double timeused = (double) (endTestTime - startTestTime) / 1000000.0;
		sb.append(String.valueOf(timeused));
		sb.append(",");
		sb.append(ParaId);
		sb.append(",");
		sb.append(",");
		sb.append((result != -1));
		sb.append(",");
		sb.append(result);
		String logString = sb.toString();
		return logString;
	}
	
	private int executeSelectStatment2(String cql)
	{
		String key = "";
		try
		{
			CassandraResultSet result = (CassandraResultSet) laudStatement.executeQuery(cql);
			int count = 0;
			if (result.next())
			{
				key = JdbcUTF8.instance.compose(ByteBuffer.wrap(result.getKey()));
				count = 0;
				if (result.getMetaData().getColumnCount() == 2)
				{
					count = result.getInt(2);
				}
			}
			System.out.println("RowKey=" + key + " Value=" + count);
			return count;
		}
		catch (SQLException e)
		{
			logger.error(e.toString());
			return -1;
		}
		catch (Exception e)
		{
			logger.error(e.toString());
			System.out.println("RowKey=" + key);
			logger.error("Error in execute cql: " + cql);
			return -1;
		}
	}
	
	private int executeSelectStatment(String cql, boolean columnCount)
	{
		String key = "";
		try
		{
			CassandraResultSet result = (CassandraResultSet) laudStatement.executeQuery(cql);
			int count = 0;
			if (columnCount)
			{
				if (result.next())
				{
					key = JdbcUTF8.instance.compose(ByteBuffer.wrap(result.getKey()));
					count = result.getMetaData().getColumnCount();
				}
			}
			else
			{
				while (result.next())
				{
					count++;
				}
			}
			return count;
		}
		catch (SQLException e)
		{
			logger.error(e.toString());
			return -1;
		}
		catch (Exception e)
		{
			logger.error(e.toString());
			System.out.println("RowKey=" + key);
			logger.error("Error in execute cql: " + cql);
			return -1;
		}
	}
	
	/**
	 * 计算两个日期的时间间隔
	 * 
	 * @param sDate开始时间
	 * @param eDate结束时间
	 * @return interval时间间隔-日
	 * */
	private static int calInterval(Date sDate, Date eDate)
	{
		int interval = 0;
		
		Calendar sCalendar = Calendar.getInstance();
		sCalendar.setTime(sDate);
		int sYears = sCalendar.get(Calendar.YEAR);
		int sDays = sCalendar.get(Calendar.DAY_OF_YEAR);
		
		Calendar eCalendar = Calendar.getInstance();
		eCalendar.setTime(eDate);
		int eYears = eCalendar.get(Calendar.YEAR);
		int eDays = eCalendar.get(Calendar.DAY_OF_YEAR);
		
		interval = 365 * (eYears - sYears);
		interval += (eDays - sDays);
		// 除去闰年天数
		while (sYears < eYears)
		{
			if (isLeapYear(sYears))
			{
				--interval;
			}
			++sYears;
		}
		// 返回计算结果
		return interval;
	}
	
	private static boolean isLeapYear(int year)
	{
		return (year % 400 == 0 || (year % 4 == 0 && year % 100 != 0));
	}
}
